// DsmServer.cc
// author: dipina@eside.deusto.es

#include "DsmServer.h"

extern "C" {
	#include </usr/include/signal.h>
}

#define SPACES " \t\r\n"

inline string trim_right (const string & s, const string & t = SPACES) { 
	string d (s); 
	string::size_type i (d.find_last_not_of (t));
	if (i == string::npos)
		return "";
	else
		return d.erase (d.find_last_not_of (t) + 1) ; 
}  // end of trim_right


inline string trim_left (const string & s, const string & t = SPACES) { 
	string d (s); 
	return d.erase (0, s.find_first_not_of (t)) ; 
}  // end of trim_left


inline string trim (const string & s, const string & t = SPACES) { 
	string d (s); 
	return trim_left (trim_right (d, t), t) ; 
}  // end of trim


inline string str(int num) {
	char buffer[32];
	sprintf(buffer, "%d", num);
	return string(buffer);
}

inline string str(unsigned long num) {
	char buffer[32];
	sprintf(buffer, "%d", num);
	return string(buffer);
}

namespace PracticaCaso {
	DsmServer::DsmServer(int p): nidCounter(-1), nodeCounter(0), TcpListener(p)
	{
		// implementado: create lock 
		pthread_rwlock_init( &lock, NULL );
	}


	DsmServer::~DsmServer()
	{
		this->stop();
		// implementado: destory lock
		pthread_rwlock_destroy( &lock );

	}

	/*
	DsmNodeId dsm_init(TcpClient *): will be invoked every time a node is
	instantiated, i.e. whenever the constructor of DsmDriver sends a dsm_init
	message to the server. The server will return the nid (node ID) which will identify
	that node in all the transactions carried out with the server, from that moment. The
	implementation of dsm_init creates a new entry in the map dsmNodeMap, where
	the metadata of every node (DsmNodeMetadata) who has initiated a session with
	the DsmServer are stored.
	*/
	DsmNodeId DsmServer::dsm_init(TcpClient * dmsClient)
	{
		DsmNodeMetadata metadata;
		pthread_rwlock_wrlock( &lock );		// ini escritura variable
		metadata.nid = ++nidCounter;
		pthread_rwlock_unlock( &lock );		// fin escritura variable
		metadata.client = dmsClient;
		pthread_rwlock_wrlock( &lock );		// ini escritura variable
		dsmNodeMap[metadata.nid] = metadata;
		nodeCounter++;
		pthread_rwlock_unlock( &lock );		// fin escritura variable
	
		return metadata.nid;
	}
	
	/*
	void dsm_exit(DsmNodeId): will be invoked everytime a DSM node is
	destroyed, i.e. whenever the DsmDriver destructor delivers a dsm_exit <nid>
	to the server to indicate that the session with DsmServer wants to be concluded.
	Notice that every node will have to send its identifier (nid) in all the messages
	exchanged with the server. The server will confirm the reception of a dsm_exit
	returning the nid or sending a message with the format ERROR: <razón>, in case
	of requesting the exit of an existing node. The implementation of dsm_exit
	eliminates the corresponding entry in map dsmNodeMap.
	*/
	void DsmServer::dsm_exit(DsmNodeId nodeId) {
		// Remove all the data structures created by this node
		pthread_rwlock_rdlock( &lock );		// ini lectura variable
		if (dsmNodeMap.find(nodeId) != dsmNodeMap.end())
		{
			pthread_rwlock_unlock( &lock );		// fin lectura variable
			pthread_rwlock_wrlock( &lock );		// ini escritura variable
			--nodeCounter;
			pthread_rwlock_unlock( &lock );		// fin escritura variable
			pthread_rwlock_rdlock( &lock );		// ini lectura variable
			if (nodeCounter == 0)
			{
				//pthread_rwlock_unlock( &lock );		// fin lectura variable
				//pthread_rwlock_rdlock( &lock );		// ini lectura variable
				for (int i=0; i<dsmNodeMap[nodeId].dsmBlocksRequested.size(); i++)
				{
					pthread_rwlock_unlock( &lock );		// fin lectura variable
					pthread_rwlock_wrlock( &lock );		// ini escritura variable
					(this->blockMetadataMap).erase(dsmNodeMap[nodeId].dsmBlocksRequested[i].blockId);
					free(dsmNodeMap[nodeId].dsmBlocksRequested[i].addr);
					pthread_rwlock_unlock( &lock );		// fin escritura variable
				}
				
			}
			else
			{
				pthread_rwlock_unlock( &lock );		// fin lectura variable
			}
			pthread_rwlock_wrlock( &lock );		// ini escritura variable
			dsmNodeMap.erase(nodeId);
			pthread_rwlock_unlock( &lock );		// fin escritura variable
		}
		else
		{
			pthread_rwlock_unlock( &lock );		// fin lectura variable
		}
	}
	
	/*
	void * dsm_malloc(DsmNodeId nid, string blockId, int
	size): will be invoked every time that a DSM node delivers a dsm_malloc
	<nid> <blockId> <size> message to the server. The server will return the
	identifier of the block created or an error message (ERROR: <razón>) as response.
	As a result of this invocation the DsmServer will update its maps dsmNodeMap
	and blockMetadataMap which register the metadata of a node (among them, the
	blocks created by it) and the metadata of a created block, respectively.
	*/
	void * DsmServer::dsm_malloc(DsmNodeId nid, string blockId, int size)
	{
		pthread_rwlock_rdlock( &lock );		// ini lectura variable
		if (this->dsmNodeMap.find(nid) != this->dsmNodeMap.end())
		{
			if (this->blockMetadataMap.find(blockId) == this->blockMetadataMap.end())
			{
				pthread_rwlock_unlock( &lock );		// fin lectura variable
				DsmBlock block;
				block.addr = malloc(size);
				if (block.addr != NULL)
				{
					block.blockSize = size;
					block.size = 0;
					block.creatorNode = nid;
					block.lastAccessNode = nid;
					
					pthread_rwlock_wrlock( &lock );		// ini escritura variable
					this->blockMetadataMap[blockId] = block;
					pthread_rwlock_unlock( &lock );		// fin escritura variable
					
					pthread_rwlock_rdlock( &lock );		// ini lectura variable
					DsmNodeMetadata metadata = this->dsmNodeMap[nid];
					pthread_rwlock_unlock( &lock );		// fin lectura variable
					
					metadata.dsmBlocksRequested.push_back(block);
					
					pthread_rwlock_wrlock( &lock );		// ini escritura variable
					this->dsmNodeMap[nid] = metadata;
					pthread_rwlock_unlock( &lock );		// fin escritura variable
					
					return block.addr;
				}
				else
				{
					cerr << "ERROR: DMS Server ran out of memory!!!" << endl;
					return 0;
				}
			}
			else
			{
				pthread_rwlock_unlock( &lock );		// fin lectura variable
				cerr << "WARNING: attempt to create block " << blockId << " already existing by " << nid << "!!!" << endl;
				pthread_rwlock_rdlock( &lock );		// ini lectura variable
				DsmBlock tempBlock = this->blockMetadataMap[blockId];
				pthread_rwlock_unlock( &lock );		// fin lectura variable
				
				if (tempBlock.size < size) {
					cerr << "ERROR: impossible to reuse block " << blockId << " of size " << tempBlock.size << " < " << size << endl;
					return 0;
				}
				else
				{
					return tempBlock.addr;
				}
			}
		}
		else
		{
			pthread_rwlock_unlock( &lock );		// fin lectura variable
			cerr << "ERROR: attempt to create block " << blockId << " by non-registered node " << nid << "!!!" << endl;
			return 0;
		}
	}

	/*
	bool dsm_put(DsmNodeId nid, string blockId, void *
	content, int size) : will be invoked every time the DsmServer receives a
	dsm_put <nid> <blockId> <size> <data> message from a node. The
	server will return to the invoking node the blockId in case of success or an
	explaining error message otherwise. As a result of this invocation, the server will
	update the corresponding entry (DsmBlock) to the block modified in the map
	blockMetadataMap.
	*/
	bool DsmServer::dsm_put(DsmNodeId nid, string blockId, void * content, int size)
	{
		pthread_rwlock_rdlock( &lock );		// ini lectura variable
		if (this->blockMetadataMap.find(blockId) != this->blockMetadataMap.end())
		{
			pthread_rwlock_unlock( &lock );		// fin lectura variable
			bool dsmPutResult = false;
			pthread_rwlock_rdlock( &lock );		// ini lectura variable
			DsmBlock blockMetadata = this->blockMetadataMap[blockId];
			//pthread_rwlock_unlock( &lock );		// fin lectura variable
			//pthread_rwlock_rdlock( &lock );		// ini lectura variable
			// We allow anybody to write over the blocks
			if ( size <= blockMetadata.blockSize )
			{
				bzero(blockMetadata.addr, blockMetadata.blockSize);
				memcpy(blockMetadata.addr, content, size);
				pthread_rwlock_unlock( &lock );		// fin lectura variable
				pthread_rwlock_wrlock( &lock );		// ini escritura variable
				blockMetadata.size = size;
				blockMetadata.lastAccessNode = nid;
				this->blockMetadataMap[blockId] = blockMetadata;
				pthread_rwlock_unlock( &lock );		// fin escritura variable
				dsmPutResult = true;
			}
			else
			{
				pthread_rwlock_unlock( &lock );		// fin lectura variable
				cerr << "ERROR: The node " << nid << " does not have write access!!!" << endl;
			}
			return dsmPutResult;
		}
		else
		{
			pthread_rwlock_unlock( &lock );		// fin lectura variable
			cerr << "ERROR: blockId " + blockId + " does not exist" << endl;
			return false;
		}
	}

	/*
	the class DsmServer offers two public methods dsm_notify_put()
	and dsm_notify_free(), which are used to send a multicast message to all
	registered DSM nodes about updates undertaken over a block identified by blockId.
	These two public methods delegate the message delivery to the private method
	dsm_notify(), which instantiates an object of type DsmNotifierThread, whose
	mission is to deliver a notification to every node registered for the multicast group
	DSM_GROUP and listening in port DSM_PORT. These constants are declared in file
	dsm/DsmTypes.h (see Figure 23). The class DsmNotifierThread avoids blocking
	the process invoking dsm_notify(), by creating a new thread to undertake the
	multicast communication.

	*/
	void DsmServer::dsm_notify(string message)
	{
		DsmNotifierThread* t = new DsmNotifierThread(message);
		t->start();
	}

	/*
		(cont.)
	*/
	void DsmServer::dsm_notify_free(int nid, string blockId)
	{
		/* now just sendto() our destination! */
		ostringstream outs;  // Declare an output string stream.
		outs << "dsm_free " << nid << " " << blockId;
		dsm_notify(outs.str());
	}

	/*
		(cont.)
	*/
	void DsmServer::dsm_notify_put(int nid, string blockId)
	{
		/* now just sendto() our destination! */
		ostringstream outs;  // Declare an output string stream.
		outs << "dsm_put " << nid << " " << blockId;
		dsm_notify(outs.str());
	}

	/*
	DsmBlock dsm_get(DsmNodeId nid, string blockId) : will be
	invoked every time the DsmServer receives a dsm_get <nid> <blockId>
	message from a node. The server will return to the invoking node the size and
	contents of the requested block in case of success or an explaining error message
	otherwise. To return such information the server simply makes a lookup of the block
	requested (blockId) in the map blockMetadataMap.
	*/
	DsmBlock DsmServer::dsm_get(DsmNodeId nid, string blockId)
	{
		pthread_rwlock_rdlock( &lock );		// ini lectura variable
		if (this->blockMetadataMap.find(blockId) != this->blockMetadataMap.end())
		{
			DsmBlock temp = this->blockMetadataMap[blockId];
			pthread_rwlock_unlock( &lock );		// fin lectura variable
			return temp;
		}
		else
		{
			pthread_rwlock_unlock( &lock );		// fin lectura variable
			DsmBlock block;
			block.blockId = "ERROR";
			return block;
		}
	}

	/*
	bool dsm_free(DsmNodeId nid, string blockId): will be invoked
	every time the DsmServer receives a dsm_free <nid> <blockId> message
	from a node. The server will return the identifier of the freed block in case of success
	or the customary error message otherwise. The corresponding entry in map
	blockMetadataMap is eliminated and the entry corresponding to the creator of
	that block in map dsmNodeMap.
	*/
	bool DsmServer::dsm_free(DsmNodeId nid, string blockId)
	{
		pthread_rwlock_rdlock( &lock );		// ini lectura variable
		if (this->dsmNodeMap.find(nid) != this->dsmNodeMap.end())
		{
			DsmNodeMetadata nodeMetadata = this->dsmNodeMap[nid];
			//pthread_rwlock_unlock( &lock );		// fin lectura variable
			//pthread_rwlock_rdlock( &lock );		// ini lectura variable
			if (this->blockMetadataMap.find(blockId) != this->blockMetadataMap.end())
			{
				DsmBlock blockMetadata = this->blockMetadataMap[blockId];
				// Only the last dsm client node who put some data can then release it
				//pthread_rwlock_unlock( &lock );		// fin lectura variable
				//pthread_rwlock_rdlock( &lock );		// ini lectura variable
				if (blockMetadata.lastAccessNode == nid)
				{
					pthread_rwlock_unlock( &lock );		// fin lectura variable
					pthread_rwlock_wrlock( &lock );		// ini escritura variable
					(this->blockMetadataMap).erase(blockId);
					pthread_rwlock_unlock( &lock );		// fin escritura variable
					pthread_rwlock_rdlock( &lock );		// ini lectura variable
					vector<DsmBlock> blocksRequested = (this->dsmNodeMap[nid]).dsmBlocksRequested;
					pthread_rwlock_unlock( &lock );		// fin lectura variable
					for (vector<DsmBlock>::iterator it = blocksRequested.begin(); it!=blocksRequested.end(); ++it)
					{
						pthread_rwlock_rdlock( &lock );		// ini lectura variable
						if ( (blockMetadata.lastAccessNode == it->lastAccessNode) && 
							 (blockMetadata.blockSize == it->blockSize) && 
							 (blockMetadata.addr == it->addr)
							)
						{
							pthread_rwlock_unlock( &lock );		// fin lectura variable
							blocksRequested.erase(it, ++it);
							pthread_rwlock_wrlock( &lock );		// ini escritura variable
							(this->dsmNodeMap[nid]).dsmBlocksRequested = blocksRequested;
							pthread_rwlock_unlock( &lock );		// fin escritura variable
							return true;
						}
						else
						{
							pthread_rwlock_unlock( &lock );		// fin lectura variable
						}
					}
					return false;
				}
				else
				{
					pthread_rwlock_unlock( &lock );		// fin lectura variable
					return false;
				}
			}
			else
			{
				pthread_rwlock_unlock( &lock );		// fin lectura variable
			}
		}
		else
		{
			pthread_rwlock_unlock( &lock );		// fin lectura variable
			return false;
		}
	}

	ostream & operator << (ostream &os, DsmServer &n)
	{
		os << "lastNid: " << n.lastNid << " - nodeCounter: " << n.nodeCounter << endl;
		// << " - dsmNodeMap: " << n.dsmNodeMap << " - blockMetadataMap " << n.blockMetadataMap << endl;
		return os;
	}

	DsmServerThread::~DsmServerThread()
	{
		delete this->dsmClient;
	}

	void DsmServerThread::run()
	{
		char* dsmCmd;
		int bytesRead;
		bool dsm_exit = false;
		while (!dsm_exit) {
			bytesRead = (this->dsmClient)->receive(dsmCmd);
			if (strncmp(dsmCmd, "dsm_init", strlen("dsm_init")) == 0) {
				// DsmNodeId dsm_init(TcpClient *);
				DsmNodeId nid = (this->dsmServer).dsm_init(this->dsmClient);
				(this->dsmClient)->send(str(nid));
				cout << "dsm_init " << nid << endl;
			} else if (strncmp(dsmCmd, "dsm_exit", strlen("dsm_exit")) == 0) {
				//void dsm_exit(DsmNodeId);
				string dsmCmd2(dsmCmd, 0, bytesRead);
				DsmNodeId nid = atoi((trim(dsmCmd2.substr(strlen("dsm_exit")))).c_str());
				(this->dsmServer).dsm_exit(nid);
				(this->dsmClient)->send(str(nid));
				dsm_exit = true;
				cout << "dsm_exit " << nid << endl;
			} else if (strncmp(dsmCmd, "dsm_malloc", strlen("dsm_malloc")) == 0) {
				// void * dsm_malloc(DsmNodeId nid, string blockId, int size);
				// dsm_malloc nid size
				string dsmCmd2(dsmCmd, 0, bytesRead);
				char blockId[256];
				DsmNodeId nid;
				int size;
				sscanf (dsmCmd2.c_str(),"%*s %d %s %d", &nid, &blockId, &size);
				void *p = (this->dsmServer).dsm_malloc(nid, blockId, size);
				if (p) {
					(this->dsmClient)->send(str((unsigned long)p));
					cout << "dsm_malloc " << nid << " " << blockId << " " << size << endl;
				} else {
					(this->dsmClient)->send("ERROR in dsm_malloc: nid " + str(nid) + " has not registered with DSM server or blockId " + blockId + " already exists");
				}
		
			} else if (strncmp(dsmCmd, "dsm_put", strlen("dsm_put")) == 0) {
				// void dsm_put(DsmNodeId nid, string blockId, void * content, int size);
				// dsm_put nid addr size datum
				// dsm_put 1 12134 45 hola estudiantes caso
				int beginDataIndex = 0;
				char seps[]   = " ";
				string token = strtok(dsmCmd, seps );
				beginDataIndex += (token.size() + 1);
				token = strtok(0, seps);
				beginDataIndex += (token.size() + 1);
				int nid = atoi(token.c_str());
				string blockId = strtok(0, seps);
				beginDataIndex += (blockId.size() + 1);
				token = strtok(0, seps);
				beginDataIndex += (token.size() + 1);
				int size = atoi(token.c_str());
				// (dsmCmd + beginDataIndex)
				if ((this->dsmServer).dsm_put(nid, blockId, (void *)&dsmCmd[beginDataIndex], bytesRead-beginDataIndex)) {
					(this->dsmClient)->send(blockId);
					(this->dsmServer).dsm_notify_put(nid, blockId);
					cout << "dsm_put " << nid << " " << blockId << " " << size << " <data>" << endl;
				} else {
					string errMsg = "ERROR in dsm_put: either blockId " + blockId + " already exists or nid " + str(nid) + " does not exist";
					(this->dsmClient)->send(errMsg);
					cout << errMsg << endl;
				}
			} else if (strncmp(dsmCmd, "dsm_get", strlen("dsm_get")) == 0) {
				string dsmCmd2(dsmCmd, 0, bytesRead);
				// DsmBlock dsm_get(DsmNodeId nid, void *addr); 
				int nid;
				string blockId;
				string cmd;
				istringstream ins; // Declare an input string stream.
				ins.str(dsmCmd2);        // Specify string to read.
				ins >> cmd >> nid >> blockId;     // Reads the integers from the string.
				DsmBlock block = (this->dsmServer).dsm_get(nid, blockId);
				if (block.blockId != "ERROR") {
					(this->dsmClient)->send((char *)block.addr, block.size);
					cout << "dsm_get " << nid << " " << blockId << endl;
				} else {
					(this->dsmClient)->send("ERROR in dsm_get: blockId " + blockId + " passed does not exist"); 
				}
				cout << "dsm_get " << nid << " " << blockId << endl;
			} else if (strncmp(dsmCmd, "dsm_free", strlen("dsm_free")) == 0) {
				// bool dsm_free(DsmNodeId nid, void* addr);
				string dsmCmd2(dsmCmd, 0, bytesRead);
				DsmNodeId nid;
				string cmd, blockId;
				istringstream ins; // Declare an input string stream.
				ins.str(dsmCmd2);        // Specify string to read.
				ins >> cmd >> nid >> blockId;     // Reads the integers from the string.
				bool blockFreed = (this->dsmServer).dsm_free(nid, blockId);
				if (blockFreed) {
					(this->dsmClient)->send(blockId);
					(this->dsmServer).dsm_notify_free(nid, blockId);
					cout << "dsm_free " << nid << " " << blockId << endl;
				} else {
					string msgError = "ERROR in dsm_free(" + str(nid) + ", " + blockId + "): node has not registered or it has not created the blockId passed";
					(this->dsmClient)->send(msgError);
					cerr << msgError << endl;
				}
			}  else {
				// Invalid command received
				(this->dsmClient)->send("ERROR: unrecognized command received");
			}
			delete []dsmCmd;
		}
	}

	
	DsmNotifierThread::DsmNotifierThread(string m): message(m) {
		/* create what looks like an ordinary UDP socket */
		if ((fd=socket(AF_INET,SOCK_DGRAM,0)) < 0) {
			cerr << "Problems creating multicast socket" << endl;
			exit(1);
		}

		/* set up destination address */
		memset(&addr,0,sizeof(addr));
		addr.sin_family=AF_INET;
		addr.sin_addr.s_addr=inet_addr(DSM_GROUP);
		addr.sin_port=htons(DSM_PORT);
	}

	DsmNotifierThread::~DsmNotifierThread() {
	}

	void DsmNotifierThread::run() {
		/* now just sendto() our destination! */
		if (sendto(fd, message.c_str(),message.size(),0,(struct sockaddr *) &addr, sizeof(addr)) < 0) {
			cerr << "ERROR: sendto in DsmNotifierThread::run()" << endl;
			perror("hola");
			exit(1);
		}
	}

}


// global variable
PracticaCaso::DsmServer * dsmServer_pointer;

// function called when CTRL-C is pressed
void ctrl_c(int)
{
    printf("\nCTRL-C was pressed...\n");
	dsmServer_pointer->stop();

}


void usage() {
	cout << "Usage: DsmServer <listening-port>" << endl;
	exit(1);
}


int main(int argc, char** argv)
{
	signal(SIGINT,ctrl_c);

	if (argc != 2) {
		usage();
	}

	PracticaCaso::DsmServer dsmServer(atoi(argv[1]));
	dsmServer_pointer = &dsmServer;
	dsmServer.start();

	while (true) {
		PracticaCaso::TcpClient *client = dsmServer.acceptTcpClient();
		cout << "Tcpclient accepted: " << client << endl;
		PracticaCaso::DsmServerThread* t = new PracticaCaso::DsmServerThread(client, dsmServer);
		t->start();
	}

	dsmServer.stop();
}
